#include "gx_threadpool.h"
#include "gx_err.h"
#include <stdlib.h>
#include <stdio.h>

static void* start_threadpool(void* arg)
{
	THREAD_POOL *thpool = (THREAD_POOL*)arg;
//	pthread_detach(pthread_self());

	while(1)
	{
		pthread_mutex_lock(&thpool->mutex);

		while( queue_empty(thpool->queue) && thpool->keepworking)
		{
			if(pthread_cond_wait(&thpool->cond,&thpool->mutex) != 0)
			{
				fatal(SYSTEM_ERROR,"pthread_cond_wait");
			}
		}

		if(!thpool->keepworking)
		{
			pthread_mutex_unlock(&thpool->mutex);
			return NULL;
		}
		else
		{
			TASK *task = (TASK*)queue_out(thpool->queue);
			pthread_mutex_unlock(&thpool->mutex);
			task->function(task->arg);
			free(task);
		}
	}
	return NULL;
}


THREAD_POOL* threadpool_init(int nthreads)
{
	if(nthreads <= 0)
	{
		fatal(PARAMETER_ERROR,"number of threads less than zero\n");
	}

	THREAD_POOL *thpool = (THREAD_POOL*)malloc(sizeof(THREAD_POOL));

	if(thpool == NULL)
		fatal(NOMEM,"malloc error\n");

	thpool->thread = (pthread_t*)malloc(sizeof(pthread_t)*nthreads);
	thpool->nthreads=nthreads;
	thpool->queue = queue_new();

	if(pthread_cond_init(&thpool->cond,NULL) != 0)
	{
		fatal(SYSTEM_ERROR,"pthread_cond_init error");
	}

	if(pthread_mutex_init(&thpool->mutex,NULL) != 0)
	{
		fatal(SYSTEM_ERROR,"pthread_mutex_init error");
	}

	pthread_mutex_lock(&thpool->mutex);
	thpool->keepworking = 1;
	pthread_mutex_unlock(&thpool->mutex);

	for(int i=0;i<thpool->nthreads;i++)
	{
		pthread_create(thpool->thread+i,NULL,start_threadpool,thpool);
	}

	return thpool;
}

void threadpool_add(THREAD_POOL *thpool,void*(*function)(void* arg),void *arg)
{
	TASK* task = (TASK*)malloc(sizeof(TASK));

	if(task == NULL)
		fatal(NOMEM,"malloc error");
	task->function = function;
	task->arg = arg;
	pthread_mutex_lock(&thpool->mutex);
	queue_in(thpool->queue,task);
	pthread_mutex_unlock(&thpool->mutex);
	pthread_cond_signal(&thpool->cond);
}

void threadpool_destroy(THREAD_POOL *thpool)
{
//	pthread_rwlock_wrlock(&thpool->work);
//	pthread_rwlock_unlock(&thpool->work);

	/*
	pthread_cond_broadcast(&thpool->cond);
	pthread_cond_broadcast(&thpool->cond);
	pthread_cond_broadcast(&thpool->cond);
	pthread_cond_broadcast(&thpool->cond);
	pthread_cond_broadcast(&thpool->cond);
	pthread_cond_broadcast(&thpool->cond);
	pthread_cond_broadcast(&thpool->cond);

	for(int i=0;i<thpool->nthreads;i++)
	{
		pthread_cond_signal(&thpool->cond);
	}
	*/
	pthread_mutex_lock(&thpool->mutex);
	TASK* task = (TASK*)malloc(sizeof(TASK));
	task->function = NULL;
	task->arg = NULL;
	queue_in(thpool->queue,task);
	thpool->keepworking = 0;
	pthread_mutex_unlock(&thpool->mutex);
	pthread_cond_broadcast(&thpool->cond);

	
	for(int i=0;i<thpool->nthreads;i++)
	{

		pthread_join(*(thpool->thread+i),NULL);
		//printf("destroy:%lu\n",*(thpool->thread+i));
	}
	//printf("join end\n");
	
	
	free(thpool->thread);
	queue_freeall(thpool->queue,free);
	pthread_cond_destroy(&thpool->cond);
	pthread_mutex_destroy(&thpool->mutex);
	free(thpool);
}
